package com.gitee.ywj1352.zk;

import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.util.List;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

/**
 * @author wenjie.yang on 2019/6/18.
 */
@Slf4j
public class DistributedLock implements Lock, Watcher {

    private ZooKeeper zk = null;
    private String ROOT_LOCK = "/locks";//定义更节点
    private String WAIT_LOCK;//等待前一个锁
    private String CURRENT_LOCK; //表示当前锁

    private CountDownLatch countDownLatch;

    public DistributedLock() {
        try {
            zk = new ZooKeeper("127.0.0.1:2181", 40000000, this);//把监听器(this) 传给构造函数
            //判断更节点是否存在
            Stat stat = zk.exists(ROOT_LOCK, false);//false 表示 次事件不需要监听
            if (stat == null) {
                zk.create(ROOT_LOCK, "0".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);//创建永久节点
            }
        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }
    }

    public synchronized boolean tryLock() {
        try {
            //创建有序节点
            CURRENT_LOCK = zk.create(ROOT_LOCK + "/", "0".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);//创建零时的 有序节点
            System.out.println(Thread.currentThread().getName() + "->" + CURRENT_LOCK + ", 尝试竞争");
            //尝试竞争
            List<String> children = zk.getChildren(ROOT_LOCK, false);
            TreeSet<String> sortedSet = new TreeSet<>();
            children.forEach(s -> sortedSet.add(ROOT_LOCK + "/" + s));
            //获取当前的最小的节点
            String firstNode = sortedSet.first();
            SortedSet<String> lessThenMe = sortedSet.headSet(CURRENT_LOCK);
            if (CURRENT_LOCK.equals(firstNode)) {//当前锁是最小的锁
                //当前节点和最小节点比较 如果想当则获得锁成功
                return true;
            }
            if (!lessThenMe.isEmpty()) {//把比自己小的lock当成 等待锁
                WAIT_LOCK = lessThenMe.last();// 把比当前更小的节点 给WAIT_LOCK 让自己监听
            }
            return false;
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        return false;
    }

    public void lock() {
        if (this.tryLock()) {
            System.out.println(Thread.currentThread().getName() + "->" + CURRENT_LOCK + ", 获得锁成功");
            return;
        }
        waitForLock(WAIT_LOCK);
    }


    private boolean waitForLock(String prev) {
        //需要监听上一个节点
        try {
            Stat stat = zk.exists(prev, true);//监听当前节点的上一个
            if (stat != null) {
                System.out.println(Thread.currentThread().getName() + "->等待" + ROOT_LOCK + "/" + prev + ", 释放");
                countDownLatch = new CountDownLatch(1);
                countDownLatch.await();
                System.out.println(Thread.currentThread().getName() + "->获得锁成功");
            }
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return true;
    }


    public void lockInterruptibly() throws InterruptedException {

    }

    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
        return false;
    }

    public void unlock() {
        System.out.println(Thread.currentThread().getName() + "->等待" + CURRENT_LOCK + "释放锁");
        try {
            zk.delete(CURRENT_LOCK, -1);
            CURRENT_LOCK = null;
            zk.close();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }
    }

    public Condition newCondition() {
        return null;
    }

    //处理监听
    public void process(WatchedEvent watchedEvent) {
        if (this.countDownLatch != null) {
            countDownLatch.countDown();
        }
    }

    public static void main(String[] args) throws Exception {

        DistributedLock distributedLock = new DistributedLock();

        distributedLock.lock();


    }
}
